package com.burt.flink.demo;

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011;
import org.apache.flink.util.Collector;

import com.alibaba.fastjson.JSON;
import com.burt.flink.model.Student;
import com.burt.flink.sink.SinkToMysql;
import com.burt.flink.util.KafkaUtils;
import com.google.common.collect.Lists;

import lombok.extern.slf4j.Slf4j;

/**
 * <p>
 * 从kafka读取数据，落地到mysql
 * </p>
 *
 * @author Burt
 * @since 2019/9/8 0008 13:17
 */
@Slf4j
public class KafkaToMysql {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        Properties props = new Properties();
        props.put("bootstrap.servers", KafkaUtils.BROKER_LIST);
        props.put("zookeeper.connect", "c-server-4:2181");
        props.put("group.id", "metric-group");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("auto.offset.reset", "latest");

        SingleOutputStreamOperator<Student> student = env
                .addSource(new FlinkKafkaConsumer011<>(KafkaUtils.TOPIC, new SimpleStringSchema(), props))
                .setParallelism(1).map(value -> JSON.parseObject(value, Student.class));

        student.timeWindowAll(Time.minutes(1)).apply(new AllWindowFunction<Student, List<Student>, TimeWindow>() {
            @Override
            public void apply(TimeWindow timeWindow, Iterable<Student> iterable, Collector<List<Student>> collector)
                    throws Exception {
                ArrayList<Student> students = Lists.newArrayList(iterable);
                if (!students.isEmpty()) {
                    System.out.println("1 分钟内收集到 student 的数据条数是：" + students.size());
                    collector.collect(students);
                }
            }
        }).addSink(new SinkToMysql());
        env.execute("flink learning connectors kafka");
    }
}
